package com.zyb.dfs.client;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zyb.dfs.namenode.rpc.model.*;
import com.zyb.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;

/**
 * 文件系统给客户端实现类
 */
public class FileSystemImpl implements FileSystem {

    private static final String NAMENODE_HOSTNAME = "localhost";
    private static final Integer NAMENODE_PORT = 50070;

    private NameNodeServiceGrpc.NameNodeServiceBlockingStub namenode;
    private NIOClient nioClient;

    public FileSystemImpl() {
        ManagedChannel channel = NettyChannelBuilder
                .forAddress(NAMENODE_HOSTNAME, NAMENODE_PORT)
                .negotiationType(NegotiationType.PLAINTEXT)
                .build();
        this.namenode = NameNodeServiceGrpc.newBlockingStub(channel);
        this.nioClient = new NIOClient();
    }

    /**
     * 创建目录
     *
     * @param path 所创建的目录对应的路径
     * @throws Exception
     */
    @Override
    public void mkdir(String path) throws Exception {
        MkdirRequest request = MkdirRequest.newBuilder()
                .setPath(path)
                .build();
        MkdirResponse response = namenode.mkdir(request);
        System.out.println("创建目录响应：" + response.getStatus());
    }

    /**
     * 优雅关闭
     *
     * @throws Exception
     */
    @Override
    public void shutdown() throws Exception {
        ShutdownRequest request = ShutdownRequest.newBuilder()
                .setCode(1)
                .build();
        namenode.shutdown(request);
    }

    /**
     * 上传文件
     *
     * @param file     文件字节数组
     * @param filename 文件名
     * @throws Exception
     */
    @Override
    public Boolean upload(byte[] file, String filename, long fileSize,
                          ResponseCallback callback) throws Exception {
        // 必须先用filename发送一个RPC接口调用到master节点
        // 去尝试在文件目录树里创建一个文件
        // 此时还需要进行查重，如果这个文件已经存在，就不让你上传了
        if (!createFile(filename)) {
            return false;
        }
        System.out.println("在文件目录树中成功创建该文件......");

        // /image/product/iphone.jpg
        // 希望的是每个数据节点在物理存储的时候，其实就是会在DATA_DIR下面去建立
        // F:\\development\\tmp\\image\product，目录
        // 在目录里就有一个iphone.jpg

        // 就是找master节点去要多个数据节点的地址
        // 就是你要考虑自己上传几个副本，找对应副本数量的数据节点的地址
        // 尽可能在分配数据节点的时候，保证每个数据节点放的数据量是比较均衡的
        // 保证集群里各个机器上放的数据比较均衡
        String datanodesJson = allocateDataNodes(filename, fileSize);
        System.out.println("申请分配了2个数据节点：" + datanodesJson);

        // 依次把文件的副本上传到各个数据节点上去
        // 还要考虑到，如果上传的过程中，某个数据节点他上传失败
        // 此时你需要有一个容错机制的考量
        JSONArray datanodes = JSONArray.parseArray(datanodesJson);
        for (int i = 0; i < datanodes.size(); i++) {
            JSONObject datanode = datanodes.getJSONObject(i);
            String hostname = datanode.getString("hostname");
            String ip = datanode.getString("ip");
            int nioPort = datanode.getIntValue("nioPort");

            if (!nioClient.sendFile(hostname, nioPort, file, filename, fileSize, callback)) {
                datanode = JSONObject.parseObject(reallocateDataNode(
                        hostname + "-" + ip, filename, fileSize));
                hostname = datanode.getString("hostname");
                nioPort = datanode.getIntValue("nioPort");
                nioClient.sendFile(hostname, nioPort, file, filename, fileSize, null);
            }
        }

        return true;
    }

    @Override
    public byte[] download(String filename) throws Exception {
        // 第一个步骤，一定是调用NameNode的接口，获取这个文件的某个副本所在的DataNode
        JSONObject datanode = getDataNodeForFile(filename, null);

        // 第二个步骤，打开一个针对那个DataNode的网络连接，发送文件名过去
        // 第三个步骤，尝试从连接中读取对方传输过来的文件
        // 第四个步骤，读取到文件之后不需要写入本地的磁盘中，而是转换为一个字节数组返回即可
        String hostname = datanode.getString("hostname");
        String ip = datanode.getString("ip");
        Integer nioPort = datanode.getInteger("nioPort");
        byte[] file = null;

        try {
            file = nioClient.readFile(hostname, nioPort, filename, true);
        } catch (Exception e) {
            datanode = getDataNodeForFile(filename, ip + "-" + hostname);
            hostname = datanode.getString("hostname");
            nioPort = datanode.getInteger("nioPort");

            try {
                file = nioClient.readFile(hostname, nioPort, filename, false);
            } catch (Exception e2) {
                throw e2;
            }
        }

        return file;
    }

    /**
     * 获取文件的某个副本所在的机器
     *
     * @param filename
     * @return
     * @throws Exception
     */
    private JSONObject getDataNodeForFile(String filename, String excludedDataNodeId) throws Exception {
        GetDataNodeForFileRequest request = GetDataNodeForFileRequest.newBuilder()
                .setFilename(filename)
                .setExcludedDataNodeId(excludedDataNodeId)
                .build();
        GetDataNodeForFileResponse response = namenode.getDataNodeForFile(request);
        return JSONObject.parseObject(response.getDatanodeInfo());
    }

    private Boolean createFile(String filename) {
        CreateFileRequest request = CreateFileRequest.newBuilder()
                .setFilename(filename)
                .build();
        CreateFileResponse createFileResponse = namenode.createFile(request);
        if (createFileResponse.getStatus() == 1) {
            return true;
        }
        return false;
    }

    /**
     * 分配双副本对应的数据节点
     *
     * @param filename
     * @param fileSize
     * @return
     */
    private String allocateDataNodes(String filename, long fileSize) {
        AllocateDataNodesRequest request = AllocateDataNodesRequest.newBuilder()
                .setFilename(filename)
                .setFileSize(fileSize)
                .build();

        AllocateDataNodesResponse response = namenode.allocateDataNodes(request);
        return response.getDatanodes();
    }

    /**
     * 分配双副本对应的数据节点
     *
     * @param filename
     * @param fileSize
     * @return
     */
    public String reallocateDataNode(String excludedDataNodeId, String filename, long fileSize) {
        ReallocateDataNodeRequest request = ReallocateDataNodeRequest.newBuilder()
                .setFilename(filename)
                .setFileSize(fileSize)
                .setExcludedDataNodeId(excludedDataNodeId)
                .build();
        ReallocateDataNodeResponse response = namenode.reallocateDataNode(request);
        return response.getDatanodeInfo();
    }
}
